来源：ververica.cn

作者 | Fabian Hueske

翻译 | 王柯凝  校对 | 邱从贤（山智）

Via：https://flink.apache.org/2019/06/26/broadcast-state.html 自版本 Flink 1.5.0 以来，Apache Flink 提供了一种新的状态类型，称为广播状态（Broadcast State）。在本文中，将解释什么是广播状态，并通过示例演示如何将广播状态应用在评估基于事件流的动态模式的应用程序，并指导大家学习广播状态的处理步骤和相关源码，以便在今后的实践中能实现此类的应用。

广播状态可以用于通过一个特定的方式来组合并共同处理两个事件流。第一个流的事件被广播到另一个 operator 的所有并发实例，这些事件将被保存为状态。另一个流的事件不会被广播，而是发送给同一个 operator 的各个实例，并与广播流的事件一起处理。广播状态非常适合两个流中一个吞吐大，一个吞吐小，或者需要动态修改处理逻辑的情况。我们将使用后者的一个具体实例来解释广播状态，并在本文的其余部分里对详细的 API 加以说明。

假设电子商务类型的网站获取了所有用户的操作行为数据作为用户的操作流，网站的运营团队致力于分析用户的操作，来提高销售额，改善用户体验，并监测和预防恶意行为。网站期望实现一个流应用程序，用于检测用户事件流中的模式，但需要避免在每次模式有变化的时候还要修改和重新部署应用程序，因此我们使用另外一个特征流来读取、更新当前特征，接下来我们通过一个实例逐步阐述如何通过 Apache Flink 中的广播状态来完成相应工作。

实例的程序获取两个数据流，第一个流提供了网站上的用户操作行为数据，如上图左上方所示，一个用户的交互事件由操作的类型（用户登录、用户注销、添加到购物车或者完成付款等）和用户的 ID（按颜色编码的）组成。图中的用户操作事件流包含用户 1001 的“登出”操作，然后是用户 1003 的“支付完成”事件，以及用户 1002 的“添加到购物车”操作。

第二个流提供了应用程序要评估的用户操作模式，模式是由两个连续的操作组成的。在上图中，模式流包含了以下两种:

这样的模式有助于企业更好地分析用户行为、检测恶意行为和提高网站体验。例如，如果只是将商品添加到购物车里而没有完成后续的支付，那么网站可以采取合适的方法，更好地了解用户没有购买的原因，并采取一定的措施以提高网站的购买转化率（例如提供优惠券、免运费等）。

在上图右侧，显示了一个 operator 的三个并发实例，这些实例获取模式和用户操作行为的数据流，评估数据流上的模式，并向下游发出模式匹配事件。为了简便起见，我们的例子中的 operator 只对一个进行两次后续操作行为的模式进行评估。当从模式流中获取到新模式的时候，将替换当前活动的模式。原则上，该 operator 也可以实现评估更复杂的模式或多个模式，这些模式可以单独添加或是删除。

我们将描述负责模式匹配的程序如何处理用户的操作和模式流。

首先，向 operator 发送一个模式，该模式被广播给这个 operator 的三个并发实例，接着，每个并发实例将模式存储在广播状态中，由于广播状态只能使用广播数据来进行更新，因此所有并发实例的状态都应该是相同的。

接下来，第一个用户信息流会基于用户 ID 进行划分，并发送给 operator 的实例，分区会确保同一用户的所有操作都由同一并发实例处理。上图显示了在 operator 实例处理了第一个模式和前三个操作行为事件之后应用程序的状态。

当任务接收到新的用户操作数据时，它通过查看用户最新的和历史的操作记录来评估当前的活动模式。对于每个用户，operator 都在 keyed state 中存储用户的上一个操作。到目前为止，由于上图中的任务只为每个用户接收一个操作（我们刚刚启动了应用程序），因此不需要评估模式。最后，keyed state 中用户的上一个操作将更新为最新的操作，以便在同一用户的下一个操作行为到达时能够进行查找。

在前三个操作行为被处理了之后，下一个事件，即用户 1001 的注销操作，将被发送到处理用户 1001 的并发实例中。当并发实例接收到用户操作的数据时，它从广播状态和用户 1001 的上一个操作中查找当前的模式。由于这两个操作符合模式匹配，因此会往下游发送匹配事件。最后，该任务会通过使用最新的操作来覆盖前一个事件以更新其 keyed state。

当一个新模式进入了模式流，它会被广播给所有任务，并且每个并发实例通过使用新模式替换当前模式来更新其广播状态。

一旦广播状态更新为新模式，那么匹配逻辑将像以前一样继续执行，即用户操作行为事件按键（key）进行分区，并由负责的并发实例进行评估。

到目前为止，我们在概念上讨论了应用程序，并解释了如何使用广播状态来评估事件流上的动态模式。接下来，我们将展示如何使用 Flink 的 DataStream API 和广播状态功能实现该实例的程序代码。

让我们从程序的输入数据开始。有两个数据流：操作行为流和模式流，在这一点上，我们并不关心数据流从何而来，这些流可以从 Apache Kafka、Kinesis 或任何其它系统中获取。

Action 和 Pattern 都是 POJO，每个都含有两个字段：

作为第一步，我们将 userId 作为操作行为流上的键：

接下来，我们准备广播状态，广播状态通常表示为 MapState，这是 Flink 提供的最通用的状态接口类。

由于这个应用程序一次只评估和存储一个 Pattern，所以我们将广播状态配置成具有键类型 Void 和值类型 Pattern 的 MapState。MapState 的键永远为 null。

以 MapStateDescriptor 为参数，调用模式流上的 Broadcast 转换操作，得到一个  BroadcastStream 对象 bcedPatterns。

在获得了 keyedstreamactionsByUser 和广播流 bcedPatterns 之后，我们对两个流使用了 connect() 方法，并在连接的流上调用了 PatternEvaluator 类（见下面 PatternEvaluator 的代码）。PatternEvaluator 是实现 KeyedBroadcastProcessFunction 接口的自定义类。它调用了我们之前讨论过的模式匹配逻辑，并发出 Tuple2 的记录，其中包含用户 ID 和匹配的模式。

KeyedBroadcastProcessFunction 接口提供了三种方法来处理数据记录和发出的结果：

你可能注意到了 KeyedBroadcastProcessFunction 类方法的上下文对象，提供了对其它功能的访问方法，例如：

在本文中，我们通过学习一个应用程序的实例，来解释 Apache Flink 的广播状态是什么，以及如何应用它来评估事件流上的动态模式，除此之外本文还讨论了广播状态的 API，并展示了相关源代码。